package msgQueueTCP

import (
	"fmt"
	"gitee.com/kingzyt/common/lock"
	"gitee.com/kingzyt/common/log"
	"gitee.com/kingzyt/common/msgQueue"
	. "gitee.com/kingzyt/common/tcpShell"
	. "gitee.com/kingzyt/common/util"
	"net"
	"sync"
	"time"
)

const (
	mq_overdue_time                      = time.Minute * 10
	default_mq_long_time_no_data         = time.Second * 10
	short_time_define_from_open_to_close = time.Second * 3
)

const Heartbeat_mark_value uint32 = 0xffffffff

var heartbeat_mark = []uint32{Heartbeat_mark_value}

type MsgQueueTCP struct {
	*msgQueue.MsgQueue
	conn       *net.TCPConn
	openTime   time.Time // call Open time, maybe not Open OK time
	closedTime time.Time

	localAddr  string
	remoteAddr string

	timeNowRough  time.Time
	lastAliveTime time.Time
	keepAliveTime time.Duration
	heartbeat     bool

	quitSign chan byte
	quited   bool

	locker *lock.Lock // 用来保护conn和closedTime的操作原子性，不保护MsgQueue本身

	// datas are determined by outside, not directly controled by mq //////////////////
	UserDataUint32 uint32       // come so far, only used by client<->ssr mq
	UserDataClosed bool         // come so far, only used by client<->ssr mq
	UserDataMutex  sync.RWMutex // come so far, only used by client<->ssr mq

	rRate    AtomicInt32
	wRate    AtomicInt32
	rbRate   AtomicInt64
	wbRate   AtomicInt64
	rWaitCnt AtomicInt32
	wWaitCnt AtomicInt32
	//////////////////
}

type NewParams struct {
	msgQueue.NewParams
	KeepAliveTime    time.Duration
	Heartbeat        bool
	CloseLineMonitor bool
}

func New(name string, params NewParams) *MsgQueueTCP {
	keepAliveTime := params.KeepAliveTime
	if params.KeepAliveTime.Nanoseconds() <= 0 {
		keepAliveTime = default_mq_long_time_no_data
	}

	rlt := &MsgQueueTCP{
		conn:          nil,
		MsgQueue:      msgQueue.New(name, params.NewParams),
		closedTime:    time.Time{},
		keepAliveTime: keepAliveTime,
		heartbeat:     params.Heartbeat,
		quitSign:      make(chan byte, 1),
		locker:        lock.NewLock(),
	}

	if !params.CloseLineMonitor {
		/*
			10w的timer并发，测试消耗为：300M内存（主要是goroutine的硬性开销），cpu占用0~3%，消耗可以接受，timer准确性基本没有影响
			一个LR/GR上设定是承载10w玩家，10%~20%的在线率，也就是1w~2w，每一个mq占用streamwork最大3，mq除了sw外最大3，mqtcp加check1，mqhub加check1，共8，
			也就是最大goroutine有8w~16w，按一个goroutine 5k占用计算，需要400M~800M内存，基本没有问题
		*/
		go rlt.checker()
	}

	return rlt
}

func (self *MsgQueueTCP) Open(
	conn *net.TCPConn,
	procReceiveMsg func(mqtcp *MsgQueueTCP, marks []uint32, msg []byte),
	onOpenOK func(raddr string),
	// this is called just before status change from closing to closed
	onStopped func(mqtcp *MsgQueueTCP),
) (ok bool) {
	var mqOpenParams msgQueue.OpenParams
	func() {
		self.locker.Lock()
		defer self.locker.Unlock()

		if self.quited {
			panic(fmt.Errorf("mq(%s) is quited when want to open", self.GetName()))
		}

		if !self.closedCore() {
			panic(fmt.Errorf("mq(%s) is not closed when want to open", self.GetName()))
		}

		if conn == nil {
			panic(fmt.Errorf("MsgQueueTCP,%s Open: conn is nil", self.GetName()))
		}
		self.conn = conn

		now := time.Now()
		self.openTime = now
		self.closedTime = time.Time{}
		self.timeNowRough = now
		self.lastAliveTime = now

		self.localAddr, self.remoteAddr, ok = GetTCPConnAddr(conn)
		if !ok {
			self.closeCore("conn fail, cannot get local/remote addr")
			return
		}

		self.SetStopParams(msgQueue.StopParams{
			OnStoppedAfter: func() {
				log.Info(self.GetName(), log.Trace, "Stopped")
				/* 这里的Close,是为了检测异常情况.
				正常的Close的情况,一般应该是已经close掉了，这里的Close调用会直接退出不会留下痕迹,
				而若是异常退出的话,就会留下log，那原因为不明*/
				self.Close("unknown reason internal stop")

				if onStopped != nil {
					onStopped(self)
				}
			},
		})

		mqOpenParams = msgQueue.OpenParams{
			OnOpenOK: func() {
				if onOpenOK != nil {
					// 这个函数会在本goroutine的这个函数退出之前完成，是一个顺序执行的过程，中间不会发生conn的置空
					onOpenOK(self.remoteAddr)
				}
			},
			Reader: conn,
			ReadEOFProc: func() {
				self.Close("read eof")
			},
			ReadErrProc: func(err error) {
				self.Close("read err")
			},
			//ReadBeforeProc: func() {},
			//ReadAfterProc:  func() {},

			Writer: conn,
			WriteErrProc: func(err error) {
				self.Close("write err")
			},
			//WriteBeforeProc: func() {},
			//WriteAfterProc:  func() {},

			MsgProc: func(marks []uint32, msg []byte) {
				func() {
					self.locker.Lock()
					defer self.locker.Unlock()

					//self.lastAliveTime = time.Now()
					self.lastAliveTime = self.timeNowRough
				}()

				// heartbeat no proc, just update lastAliveTime
				// heartbeat flag is for sending hearbeat, not for receive heartbeat
				if len(marks) > 0 && marks[0] == Heartbeat_mark_value {
					//log.Info(self.GetName(), log.Trace, "heartbeat")
					return
				}

				//log.Info(self.GetName(), log.Trace, "proc msg, marks:%v, msgLen:%d", marks, len(msg))
				procReceiveMsg(self, marks, msg)
			},
		}
	}()

	if !ok {
		return false
	}

	// Open不放在lock内，因为本身msgQueue的Open无所谓锁保护，而且onOpenOK函数可能调用任何的msgQueueTCP的函数，会造成lock重入锁死
	if !self.MsgQueue.Open(mqOpenParams) {
		self.closeCore("open fail")
		return false
	}

	log.Info(self.GetName(), log.Trace, "Open")

	return true
}

/*
虚拟机上面,时间获取函数可能比较耗时,
我们避免每一个消息接收都调用time.Now(),这样就需要做一个定时采样时间,来粗略地记录消息到达时间

对于一个1/2check的计时周期,有3个计时点,分别记为0,1,2
在有接收heartbeat的情况下,考虑到time的不精确性,两个beat之间的间隔可能大于或小于设定的时间,
若设定1/2周期发送一次,那么最坏情况下,一次收到在1之前一点点,被记录在0,然后下一次收到略大于1/2,被记录在2,
然后2与0的差值因为计时器的不精确性,比设定周期略大的话,就会造成hearbeat失效

而对于1/3的check,有4个计时点,0,1,2,3
在有接收heartbeat的情况下,最坏0,2计时,也是大于设定周期,
但考虑到计时器的不精确性,周期的设定不能太短,最好是以分钟为单位,这样保证heartbeat的正常工作

在没有heartbeat的情况下,最坏情况下,接近1记为0,略超过3,总间隔实际小于周期,但判定为超时,
所以要考虑消息发送间隔在2/3,甚至1/2周期以内
*/
func (self *MsgQueueTCP) checker() {
	c := time.NewTicker(time.Duration(int64(self.keepAliveTime) / 3)).C
End:
	for {
		select {
		case <-self.quitSign:
			break End
		case <-c:
			//log.Info(self.GetName(), log.Trace, "mqtcp checker")
			self.checkLongTimeNoData()
			if self.heartbeat {
				self.sendHeartbeat()
			}
		}
	}
}

func (self *MsgQueueTCP) checkLongTimeNoData() {
	self.locker.Lock()
	defer self.locker.Unlock()

	now := time.Now()

	// cal op rate
	rc, wc, rb, wb := self.MsgQueue.GetRecord()
	sec := now.Sub(self.timeNowRough).Seconds()
	self.timeNowRough = now
	if sec <= 0 {
		return
	}
	self.rRate.Set(int32(float64(rc) / sec))
	self.wRate.Set(int32(float64(wc) / sec))
	self.rbRate.Set(int64(float64(rb) / sec))
	self.wbRate.Set(int64(float64(wb) / sec))
	self.MsgQueue.ResetRecord()

	rwc, wwc := self.MsgQueue.GetRWWaitCnt()
	self.rWaitCnt.Set(int32(rwc))
	self.wWaitCnt.Set(int32(wwc))

	if !self.closedCore() && now.Sub(self.lastAliveTime) > self.keepAliveTime {
		self.closeCore("long time no data")
	}
}

func (self *MsgQueueTCP) GetRWRate() (r int32, w int32, rb int64, wb int64) {
	return self.rRate.Get(), self.wRate.Get(), self.rbRate.Get(), self.wbRate.Get()
}

func (self *MsgQueueTCP) GetRWWaitCnt() (rWaitCnt int32, wWaitCnt int32) {
	return self.rWaitCnt.Get(), self.wWaitCnt.Get()
}

func (self *MsgQueueTCP) sendHeartbeat() {
	self.locker.Lock()
	defer self.locker.Unlock()

	if !self.closedCore() {
		if self.SendMsg(heartbeat_mark, nil) {
			//log.Info(self.GetName(), log.Trace, "send heartbeat")
		}
	}
}

// 内部使用，无锁版本，靠调用方来锁保护
func (self *MsgQueueTCP) closeCore(reason string) {
	if self.conn == nil {
		//log.Info(self.GetName(), log.Trace, "try Close fail: self.conn == nil")
		return
	}

	log.Info(self.GetName(), log.I, "Close: %s", reason)

	self.conn.Close()
	self.conn = nil
	self.closedTime = time.Now()

	self.MsgQueue.Close()
}

func (self *MsgQueueTCP) Close(reason string) {
	self.locker.Lock()
	defer self.locker.Unlock()

	self.closeCore(reason)
}

func (self *MsgQueueTCP) Quit() {
	self.locker.Lock()
	defer self.locker.Unlock()

	self.closeCore("quit")
	self.quitSign <- 1
	close(self.quitSign)
	self.quited = true
}

// 内部使用，无锁版本，靠调用方来锁保护
func (self *MsgQueueTCP) closedCore() bool {
	return self.conn == nil && self.MsgQueue.Closed()
}

func (self *MsgQueueTCP) Closed() bool {
	self.locker.Lock()
	defer self.locker.Unlock()

	return self.closedCore()
}

func (self *MsgQueueTCP) Overdue() bool {
	self.locker.Lock()
	defer self.locker.Unlock()

	return self.closedCore() && !self.closedTime.IsZero() && time.Now().Sub(self.closedTime) > mq_overdue_time
}

func (self *MsgQueueTCP) IsShortFromOpenToClose() bool {
	self.locker.Lock()
	defer self.locker.Unlock()

	return self.closedTime.Sub(self.openTime) < short_time_define_from_open_to_close
}

func (self *MsgQueueTCP) RemoteAddr() string {
	self.locker.Lock()
	defer self.locker.Unlock()

	return self.remoteAddr
}

func (self *MsgQueueTCP) LocalAddr() string {
	self.locker.Lock()
	defer self.locker.Unlock()

	return self.localAddr
}
